/*******************************************************************************
 * This file is part of Crunch Network.
 *
 * Crunch Network is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free Software
 * Foundation, either version 3 of the License, or (at your option) any later version.
 *
 * Crunch Network is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
 * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License along with
 * Crunch Network.  If not, see <http://www.gnu.org/licenses/>.
 ******************************************************************************/

package com.crunch.network;

import com.crunch.network.util.ExponentiallyWeightedMovingAverage;

import java.util.ArrayDeque;
import java.util.Deque;

/**
 * The {@code NetworkConnection} class provides a virtual UDP connection.
 */
public class NetworkConnection {
	/**
	 * Constructs a {@code NetworkConnection}. This method is at package scope because it should only be called by the
	 * {@link NetworkConnectionManager} instance.
	 *
	 * @param   settings the settings used to initialize this connection.
	 * @param   id the ID of this connection, determined by the {@link NetworkConnectionManager}.
	 * @param   address the address of the device on which the connection is being created.
	 * @param   timeMs the time in milliseconds this connection was created.
	 */
	NetworkConnection(NetworkSettings settings, int id, NetworkAddress address, long timeMs) {
		this.settings = settings;

		connectionId = id;
		this.address = new NetworkAddress(address);
		creationTimeMs = timeMs;
		lastReceiveTimeMs = 0;

		nextLocalSequenceNumber = 0;
		localSequenceNumberWrapAdd = 0;

		newestRemoteSequenceNumber = -1; // value -1 represents "no remote packets received"
		remoteSequenceNumberBitfield = 0;
		remoteSequenceNumberWrapAdd = 0;

		newestAck = 0;
		ackQueue = 0;
		ackQueueInit();

		pingTimeAvg = new ExponentiallyWeightedMovingAverage(settings.pingSmoothing);
		pingTimeMs = 0;
		jitterMs = 0;
		pingQueue = new ArrayDeque<Long>(20);
		packetDropRate = 0.0f;
		sentPacketSizes = new ArrayDeque<PacketSize>(20);
		sendDataRate = 0;
		receivedPacketSizes = new ArrayDeque<PacketSize>(20);
		receivedDataRate = 0;
	}

	/**
	 * Returns the connection ID used to uniquely identify this connection.
	 *
	 * @return  the connection ID.
	 */
	public int getConnectionId() {
		return connectionId;
	}

	/**
	 * Returns the address of the device on which the connection was created.
	 *
	 * @return  the device address.
	 */
	public NetworkAddress getNetworkAddress() {
		return new NetworkAddress(address);
	}

	/**
	 * Returns the time in milliseconds at which this connection was created.
	 *
	 * @return  the time at which this connection was created.
	 */
	public long getCreationTimeMs() {
		return creationTimeMs;
	}

	/**
	 * Returns the time in milliseconds at which the last packet was received on this connection.
	 *
	 * @return  time at which the last packet was received.
	 */
	public long getLastReceiveTimeMs() {
		return lastReceiveTimeMs;
	}

	/**
	 * Returns the ping time in milliseconds of this connection.
	 *
	 * @return  the ping time.
	 */
	public long getPingTimeMs() {
		return pingTimeMs;
	}

	/**
	 * Returns the jitter time in milliseconds of this connection.
	 *
	 * @return  the jitter time.
	 */
	public long getJitterMs() {
		return jitterMs;
	}

	/**
	 * Returns the rate at which packets sent from this connection are being dropped. A rate of 0.0 means that no
	 * packets are being dropped and a rate of 1.0 means that all packets are being dropped.
	 *
	 * @return  the rate at which packets are being dropped.
	 */
	public float getPacketDropRate() {
		return packetDropRate;
	}

	/**
	 * Returns the rate at which data is being sent on this connection in bytes per second.
	 *
	 * @return  the rate at which data is being sent.
	 */
	public int getSendDataRate() {
		return sendDataRate;
	}

	/**
	 * Returns the rate at which data is being received on this connection in bytes per second.
	 *
	 * @return  the rate at which data is being received.
	 */
	public int getReceiveDataRate() {
		return receivedDataRate;
	}

	/**
	 * Called when a packet is received to update the connection's internal state. This method is at package scope
	 * because it should only be called by the {@link NetworkConnectionManager}.
	 *
	 * @param   packet the packet into which the received data was stored.
	 * @param   receiveTimeMs the time at which the packet is being received.
	 */
	void receivePacket(NetworkPacket packet, long receiveTimeMs) {
		// we calculate the packet data rate even if we disregard the packet
		PacketSize ps = new PacketSize();
		ps.timeMs = receiveTimeMs;
		ps.sizeBytes = packet.getPacketDataSize();
		receivedPacketSizes.addLast(ps);
		receivedDataRate += ps.sizeBytes;

		// determine if this is the first remote packet received - used for several things later
		boolean first = newestRemoteSequenceNumber < 0;
		if (first) {
			newestRemoteSequenceNumber = 0;
		}

		// when we receive a packet, we first update the list of remote packets that we have received
		// get the sequence number of the packet
		int remoteSequenceNumber = packet.getSequenceNumber();
		// compute (newest - just_received)
		int difference = sequenceNumberDifference(newestRemoteSequenceNumber, remoteSequenceNumber);
		if (difference >= 0) {
			// the packet is older than the most recently received packet, or it's a duplicate of it
			// in this case, we just drop the packet
			return;
		}

		lastReceiveTimeMs = receiveTimeMs;

		// adjust the remote sequence number bitfield
		int shift = -difference;
		if (shift >= 32) {
			// all bits are shifted off the edge, but bit 0 becomes a 1 because it refers to remoteSequenceNumber
			remoteSequenceNumberBitfield = 1;
		} else {
			// shift bits and set bit 0 to 1 because it refers to remoteSequenceNumber
			remoteSequenceNumberBitfield = (remoteSequenceNumberBitfield << shift) | 1;
		}

		if (remoteSequenceNumber < newestRemoteSequenceNumber) {
			// if this sequence number is newer but less in value, we must have wrapped
			remoteSequenceNumberWrapAdd += NetworkPacket.MAX_SEQUENCE_NUMBER_VALUE + 1;
		}
		newestRemoteSequenceNumber = remoteSequenceNumber;

		// next, we examine the ACK list
		int packetAck = packet.getAckSequenceNumber();
		// check to make sure this ACK isn't for a packet we haven't even sent yet
		int packetAckId = getLocalPacketId(packetAck);
		int nextPacketId = getLocalPacketId(nextLocalSequenceNumber);
		if (packetAckId < nextPacketId || first) {
			// this should always be the case with valid packets

			int packetAckBitfield = packet.getAckBitfield();
			int ackDifference = sequenceNumberDifference(newestAck, packetAck);
			if ((ackDifference < 0 || ackQueueEmpty()) && (packetAckBitfield & 1) != 0) {
				// the ACK received is at least as recent as our current most recent ACK
				// this should always happen (with valid packets) since we discard out-of-order packets
				// we also make sure that the first bit of the ack bitfield is 1
				// if it isn't, this indicates that we have received no ACKs

				// we check the range [newestAck+1, packetAck]
				// if any bits in this range are 0, it means the packet was dropped
				// the exception is if the ack queue is empty - in this case we start from 0
				// (this is because if the queue is empty, this is our first packet received)
				int start = ackQueueEmpty() ? 0 : getLocalPacketId(newestAck) + 1;

				// first make sure the queue doesn't grow past 32 elements
				int newQueueSize = ackQueueSize() + (packetAckId + 1 - start);
				for (int i = 32; i < newQueueSize && !ackQueueEmpty(); ++i) {
					ackQueuePop();
				}

				for (int i = start; i <= packetAckId; ++i) {
					boolean ack;
					// determine the bit in the bitfield representing the ACK
					int ackBit = packetAckId - i;
					if (ackBit >= 32) {
						// if the bit is >= 32, the bitfield doesn't have enough space
						// in this case, we assume the packet is dropped
						ack = false;
					} else {
						// otherwise, check the ACK bit
						ack = ((packetAckBitfield & (1 << ackBit)) != 0);
						ackQueuePush(ack);
					}

					if (ack) {
						// adjust ping
						// the front of the ping queue is our packet
						long packetPingTime = receiveTimeMs - pingQueue.getFirst();
						pingTimeAvg.addDataPoint((float) packetPingTime);
						pingTimeMs = Math.round(pingTimeAvg.getAverage());
						jitterMs = Math.round((float) Math.sqrt(pingTimeAvg.getVariance()));

						// packet was acknowledged
						settings.callbacks.onPacketAcknowledged(this, packetAckId - ackBit, receiveTimeMs);
					} else {
						// packet was dropped
						settings.callbacks.onPacketDropped(this, packetAckId - ackBit, receiveTimeMs);
					}

					pingQueue.removeFirst();
				}

				// compute drop rate
				int dropped = ackQueueCount();
				packetDropRate = (float) dropped / 32.0f;

				// update the acknowledgements we've received
				newestAck = packetAck;
			}
		}

		settings.callbacks.processPacket(this, packet, receiveTimeMs);
	}

	/**
	 * Called when a packet is sent to update the connection's internal state. This method is at package scope because
	 * it should only be called by the {@link NetworkConnectionManager}.
	 *
	 * @param   packet the packet in which the sent data was stored.
	 * @param   sendTimeMs the time at which the packet is being sent.
	 */
	void sendPacket(NetworkPacket packet, long sendTimeMs) {
		// set the packet's protocol ID
		packet.setProtocolId(settings.protocolId);
		// set the packet's sequence number to the next remote sequence number
		packet.setSequenceNumber(nextLocalSequenceNumber);
		// set the packet's acknowledgement to the last remote sequence number we received, plus the bitfield of the previous 32
		packet.setAck(newestRemoteSequenceNumber, remoteSequenceNumberBitfield);

		// check if we're about to wrap around
		// if so, make sure we add to the wrap counter
		if (nextLocalSequenceNumber == NetworkPacket.MAX_SEQUENCE_NUMBER_VALUE) {
			localSequenceNumberWrapAdd += NetworkPacket.MAX_SEQUENCE_NUMBER_VALUE + 1;
		}

		// add to the ping queue
		pingQueue.addLast(sendTimeMs);

		// update send data rate
		PacketSize ps = new PacketSize();
		ps.timeMs = sendTimeMs;
		ps.sizeBytes = packet.getPacketDataSize();
		sentPacketSizes.addLast(ps);
		sendDataRate += ps.sizeBytes;

		// wraps when we go past 65535
		nextLocalSequenceNumber = (nextLocalSequenceNumber == NetworkPacket.MAX_SEQUENCE_NUMBER_VALUE) ? 0 : nextLocalSequenceNumber+1;
	}

	/**
	 * Recalculates the received/sent data rates for the given time in milliseconds.
	 *
	 * @param   timeMs the current time in milliseconds.
	 */
	void calculateDataRates(long timeMs) {
		while (!sentPacketSizes.isEmpty() && sentPacketSizes.getFirst().timeMs + 1000 < timeMs) {
			sendDataRate -= sentPacketSizes.getFirst().sizeBytes;
			sentPacketSizes.removeFirst();
		}

		while (!receivedPacketSizes.isEmpty() && receivedPacketSizes.getFirst().timeMs + 1000 < timeMs) {
			receivedDataRate -= receivedPacketSizes.getFirst().sizeBytes;
			receivedPacketSizes.removeFirst();
		}
	}

	/**
	 * Returns the unique ID of the given local {@link NetworkPacket}. A local packet is one which is being sent from
	 * the local device.
	 *
	 * @param   packet the packet for which to return the ID.
	 * @return  the ID of the packet.
	 */
	public int getLocalPacketId(NetworkPacket packet) {
		return getLocalPacketId(packet.getSequenceNumber());
	}

	/**
	 * Returns the unique ID of the given remote {@link NetworkPacket}. A remote packet is one which is being sent from
	 * a remote device.
	 *
	 * @param   packet the packet for which to return the ID.
	 * @return  the ID of the packet.
	 */
	public int getRemotePacketId(NetworkPacket packet) {
		return getRemotePacketId(packet.getSequenceNumber());
	}

	private NetworkSettings settings;   // the network settings

	private int connectionId;       // ID of the connection
	private NetworkAddress address; // the address that we are "connected" to
	private long creationTimeMs;    // the time in milliseconds this connection was created
	private long lastReceiveTimeMs; // the time in milliseconds the last packet was received

	// along with each packet, a sequence number is sent to uniquely identify the packet
	// the "absolute" sequence number is a 4-byte integer, but only the lower 2 bytes are sent on the network
	// therefore, each side of the connection manually tracks wrapping and reconstructs the upper 2 bytes

	private int nextLocalSequenceNumber;    // the next local sequence number to send
	private int localSequenceNumberWrapAdd; // the amount to add to the next local sequence number due to wrapping

	private int newestRemoteSequenceNumber;     // the newest remote sequence number received - initially -1 (meaning none received)
	private int remoteSequenceNumberBitfield;   // the remote sequence number bitfield referring to [newest-31, newest]
	private int remoteSequenceNumberWrapAdd;    // the amount to add to the newest remote sequence number due to wrapping

	// each packet received contains a set of "acknowledgements" for the last 32 packets RECEIVED by that connection
	// the acknowledgements take the form of a sequence number followed by a 32-bit bitfield
	// the ith bit represents whether packet (seq_num - i) has been received

	private int newestAck;      // the newest acknowledgement sequence number received
	private int ackQueue;       // queue of acknowledgements for the last 32 packets
	private int ackQueueFront;  // front of ackQueue
	private int ackQueueSize;   // size of ackQueue

	private void ackQueueInit() {
		ackQueueFront = 0;
		ackQueueSize = 0;
	}

	private boolean ackQueueEmpty() {
		return ackQueueSize == 0;
	}

	private int ackQueueSize() {
		return ackQueueSize;
	}

	private int ackQueueCount() {
		// count the number of bits
		int count = 0;
		for (int i = 0; i < ackQueueSize; ++i) {
			int index = (ackQueueFront + i) % 32;
			if ((ackQueue & (1 << index)) != 0) {
				++count;
			}
		}
		return count;
	}

	private void ackQueuePush(boolean v) {
		// push onto back
		int index = (ackQueueFront + ackQueueSize) % 32;
		int bitMask = 1 << index;
		ackQueue = v ? (ackQueue | bitMask) : (ackQueue & ~bitMask);
		++ackQueueSize;
	}

	private void ackQueuePop() {
		// pop from front
		ackQueueFront = (ackQueueFront + 1) % 32;
	}

	// returns a - b, taking wrapping into account
	private static int sequenceNumberDifference(int a, int b) {
		final int SEQUENCE_MAX = NetworkPacket.MAX_SEQUENCE_NUMBER_VALUE + 1;
		final int SEQUENCE_HALF = SEQUENCE_MAX / 2;

		int diff = a - b;
		if (a >= b) {
			if (diff < SEQUENCE_HALF) {
				return diff;
			} else {
				return diff - SEQUENCE_MAX;
			}
		} else {
			if (diff >= -SEQUENCE_HALF) {
				return diff;
			} else {
				return diff + SEQUENCE_MAX;
			}
		}
	}

	// computes full local packet ID using wrap add
	private int getLocalPacketId(int sequenceNumber) {
		// get the difference between the two
		int difference = sequenceNumberDifference(nextLocalSequenceNumber, sequenceNumber);
		// add back wrapping value
		return localSequenceNumberWrapAdd + (nextLocalSequenceNumber - difference);
	}

	// computes full remote packet ID using wrap add
	private int getRemotePacketId(int sequenceNumber) {
		// get the difference between the two
		int difference = sequenceNumberDifference(newestRemoteSequenceNumber, sequenceNumber);
		// add back wrapping value
		return remoteSequenceNumberWrapAdd + (newestRemoteSequenceNumber - difference);
	}

	// variables for tracking network stats

	// contains time/size data about a sent packet
	private static class PacketSize {
		public long timeMs;                                 // time packet was sent in milliseconds
		public int sizeBytes;                               // size of packet
	}
	private ExponentiallyWeightedMovingAverage pingTimeAvg; // keeps track of ping average/variance
	private int pingTimeMs;                                 // ping time in milliseconds
	private int jitterMs;                                   // jitter in milliseconds
	private Deque<Long> pingQueue;                          // queue of recently sent packet send times, used to determine ping
	private float packetDropRate;                           // rate at which packets are being dropped
	private Deque<PacketSize> sentPacketSizes;              // queue of recently sent packet sizes over the last second
	private int sendDataRate;                               // data (in bytes) sent over the last second
	private Deque<PacketSize> receivedPacketSizes;          // queue of recently received packet sizes over the last second
	private int receivedDataRate;                           // data (in bytes) received over the last second
}
